package com.ihr360.job.core.item;

import com.ihr360.job.core.StepListener;
import com.ihr360.job.core.entity.StepContribution;
import com.ihr360.job.core.job.builder.JobBuilderHelper;
import com.ihr360.job.core.listener.MulticasterBatchListener;
import com.ihr360.job.core.scope.context.SimpleStepContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import java.util.List;

/**
 * Simple implementation of the {@link ChunkProcessor} interface that handles
 * basic item writing and processing. Any exceptions encountered will be
 * rethrown.
 *
 * @see ChunkOrientedTasklet
 */
public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, InitializingBean {
    protected final Logger logger = LoggerFactory.getLogger(JobBuilderHelper.class.getName());
    private ItemProcessor<? super I, ? extends O> itemProcessor;

    private ItemWriter<? super O> itemWriter;

    private final MulticasterBatchListener<I, O> listener = new MulticasterBatchListener<I, O>();

    /**
     * Default constructor for ease of configuration (both itemWriter and
     * itemProcessor are mandatory).
     */
    @SuppressWarnings("unused")
    private SimpleChunkProcessor() {
        this(null, null);
    }

    public SimpleChunkProcessor(ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter) {
        this.itemProcessor = itemProcessor;
        this.itemWriter = itemWriter;
    }

    /**
     * @param itemProcessor the {@link ItemProcessor} to set
     */
    public void setItemProcessor(ItemProcessor<? super I, ? extends O> itemProcessor) {
        this.itemProcessor = itemProcessor;
    }

    /**
     * @param itemWriter the {@link ItemWriter} to set
     */
    public void setItemWriter(ItemWriter<? super O> itemWriter) {
        this.itemWriter = itemWriter;
    }

    /**
     * Check mandatory properties.
     *
     * @see InitializingBean#afterPropertiesSet()
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(itemWriter, "ItemWriter must be set");
        Assert.notNull(itemProcessor, "ItemProcessor must be set");
    }

    /**
     * Register some {@link StepListener}s with the handler. Each will get the
     * callbacks in the order specified at the correct stage.
     *
     * @param listeners
     */
    public void setListeners(List<? extends StepListener> listeners) {
        for (StepListener listener : listeners) {
            registerListener(listener);
        }
    }

    /**
     * Register a listener for callbacks at the appropriate stages in a process.
     *
     * @param listener a {@link StepListener}
     */
    public void registerListener(StepListener listener) {
        this.listener.register(listener);
    }

    /**
     * @return the listener
     */
    protected MulticasterBatchListener<I, O> getListener() {
        return listener;
    }

    /**
     * @param item the input item
     * @return the result of the processing
     * @throws Exception
     */
    protected final O doProcess(I item, SimpleStepContext simpleStepContext) throws Exception {

        if (itemProcessor == null) {
            @SuppressWarnings("unchecked")
            O result = (O) item;
            return result;
        }

        try {
            listener.beforeProcess(item, simpleStepContext);
            O result = itemProcessor.process(item);
            listener.afterProcess(item, result);
            return result;
        } catch (Exception e) {
            listener.onProcessError(item, simpleStepContext, e);
            throw e;
        }

    }

    /**
     * Surrounds the actual write call with listener callbacks.
     *
     * @param items
     * @throws Exception
     */
    protected final void doWrite(List<O> items) throws Exception {

        if (itemWriter == null) {
            return;
        }

        try {
            listener.beforeWrite(items);
            writeItems(items);
            doAfterWrite(items);
        } catch (Exception e) {
            doOnWriteError(e, items);
            throw e;
        }

    }

    /**
     * Call the listener's after write method.
     *
     * @param items
     */
    protected final void doAfterWrite(List<O> items) {
        listener.afterWrite(items);
    }

    protected final void doOnWriteError(Exception e, List<O> items) {
        listener.onWriteError(e, items);
    }

    protected void writeItems(List<O> items) throws Exception {
        if (itemWriter != null) {
            itemWriter.write(items);
        }
    }

    @Override
    public final void process(StepContribution contribution, Chunk<I> inputs, SimpleStepContext simpleStepContext) throws Exception {

        // Allow temporary state to be stored in the user data field
        initializeUserData(inputs);

        // If there is no input we don't have to do anything more
        if (isComplete(inputs)) {
            return;
        }

        // Make the transformation, calling remove() on the inputs iterator if
        // any items are filtered. Might throw exception and cause rollback.
        Chunk<O> outputs = transform(contribution, inputs, simpleStepContext);

        // Adjust the filter count based on available data
        contribution.incrementFilterCount(getFilterCount(inputs, outputs));

        // Adjust the outputs if necessary for housekeeping purposes, and then
        // write them out...
        write(contribution, inputs, getAdjustedOutputs(inputs, outputs));

    }

    /**
     * Extension point for subclasses to allow them to memorise the contents of
     * the inputs, in case they are needed for accounting purposes later. The
     * default implementation sets up some user data to remember the original
     * size of the inputs. If this method is overridden then some or all of
     * {@link #isComplete(Chunk)}, {@link #getFilterCount(Chunk, Chunk)} and
     * {@link #getAdjustedOutputs(Chunk, Chunk)} might also need to be, to
     * ensure that the user data is handled consistently.
     *
     * @param inputs the inputs for the process
     */
    protected void initializeUserData(Chunk<I> inputs) {
        inputs.setUserData(inputs.size());
    }

    /**
     * Extension point for subclasses to calculate the filter count. Defaults to
     * the difference between input size and output size.
     *
     * @param inputs  the inputs after transformation
     * @param outputs the outputs after transformation
     * @return the difference in sizes
     * @see #initializeUserData(Chunk)
     */
    protected int getFilterCount(Chunk<I> inputs, Chunk<O> outputs) {
        return (Integer) inputs.getUserData() - outputs.size();
    }

    /**
     * Extension point for subclasses that want to store additional data in the
     * inputs. Default just checks if inputs are empty.
     *
     * @param inputs the input chunk
     * @return true if it is empty
     * @see #initializeUserData(Chunk)
     */
    protected boolean isComplete(Chunk<I> inputs) {
        return inputs.isEmpty();
    }

    /**
     * Extension point for subclasses that want to adjust the outputs based on
     * additional saved data in the inputs. Default implementation just returns
     * the outputs unchanged.
     *
     * @param inputs  the inputs for the transformation
     * @param outputs the result of the transformation
     * @return the outputs unchanged
     * @see #initializeUserData(Chunk)
     */
    protected Chunk<O> getAdjustedOutputs(Chunk<I> inputs, Chunk<O> outputs) {
        return outputs;
    }

    /**
     * Simple implementation delegates to the {@link #doWrite(List)} method and
     * increments the write count in the contribution. Subclasses can handle
     * more complicated scenarios, e.g.with fault tolerance. If output items are
     * skipped they should be removed from the inputs as well.
     *
     * @param contribution the current step contribution
     * @param inputs       the inputs that gave rise to the outputs
     * @param outputs      the outputs to write
     * @throws Exception if there is a problem
     */
    protected void write(StepContribution contribution, Chunk<I> inputs, Chunk<O> outputs) throws Exception {
        try {
            doWrite(outputs.getItems());
        } catch (Exception e) {
            /*
             * For a simple chunk processor (no fault tolerance) we are done
			 * here, so prevent any more processing of these inputs.
			 */
            inputs.clear();
            throw e;
        }
        contribution.incrementWriteCount(outputs.size());
    }

    protected Chunk<O> transform(StepContribution contribution, Chunk<I> inputs, SimpleStepContext simpleStepContext) throws Exception {
        Chunk<O> outputs = new Chunk<O>();
        for (Chunk<I>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext(); ) {
            final I item = iterator.next();
            O output;
            try {
                output = doProcess(item, simpleStepContext);
            } catch (Exception e) {
				/*
				 * For a simple chunk processor (no fault tolerance) we are done
				 * here, so prevent any more processing of these inputs.
				 */
                inputs.clear();
                throw e;
            }
            if (output != null) {
                outputs.add(output);
            } else {
                iterator.remove();
            }
        }
        return outputs;
    }

}
